package com.hooper.iot.face.queue;

import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author :     WANGCHENG
 * @Date :     2022/6/10 9:37
 * @Version:
 * @Description 本次只做了限流功能，结合kafka本身消费提交的特性实现类似于事务不提交的功能，从而实现了超出部分的消息一直报异常并不会被消费，
 *在最初的设计中是使用queue，将每次消息存放于queue中，使用queue作为锁来控制消费，生产者代码synchronized (queue) queue.size<maxSize queue.offer()，消费者代码synchronized (queue) queue.poll()。
 */
@Component
public class ConsumerQueueReposity {
    private final static Map<String, AtomicInteger> QUEUE_MAP = new ConcurrentHashMap<>();
    public void put(String key, AtomicInteger value) {
        QUEUE_MAP.put(key, value);
    }
    public AtomicInteger get(String key) {
        synchronized (this) {
            return QUEUE_MAP.get(key);
        }
    }


    //TODO 当设备掉线时应考虑移除该设备对应的计数器，否则线程内存泄露会导致OOM。
    public void remove(String key) {
        synchronized (this) {
            QUEUE_MAP.remove(key);
        }
    }

    public int size() {
        return QUEUE_MAP.size();
    }

    public Map getMap() {
        return QUEUE_MAP;
    }
}
